package com.shujia.base;


import com.shujia.utils.MyHBaseTools;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.List;

public class HbaseFilterDemo1 {
    private Connection conn;
    private Admin admin;

    /**
     * 获取与hbase数据连接对象
     */
    @Before
    public void getConnection() {
        try {
            //创建hbase配置文件对象
//        HBaseConfiguration conf = new HBaseConfiguration(); //旧版本的使用方式
            //HBaseConfiguration中有一个静态方法create() --> Configuration对象
            Configuration conf = HBaseConfiguration.create();
            //指定要连接hbase集群(zk集群)
            conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

            //获取数据库连接对象
            //使用连接工厂类创建连接对象
            conn = ConnectionFactory.createConnection(conf);

            //创建数据库操作对象
            //新版本已经没有HBaseAdmin这个类了
            admin = conn.getAdmin();

            System.out.println("成功获取数据库连接对象: " + conn);
            System.out.println("成功获取数据库操作对象: " + admin);
            System.out.println("====================================================");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 需求1：通过RowFilter与BinaryComparator过滤比rowKey 1500100010小的所有值出来
     */
    @Test
    public void scanAndRowFilterAndBinaryComparator() {
        try {
            //将表名封装成TableName对象
            TableName tableName = TableName.valueOf("students2");

            //获取表实例对象
            Table students2 = conn.getTable(tableName);

            //创建Scan对象
            Scan scan = new Scan(); //默认情况下，查询所有数据

            //创建行键过滤器RowFilter
            //旧版本写法：RowFilter(final CompareOp rowCompareOp, final ByteArrayComparable rowComparator)
            //创建BinaryComparator比较器对象
            BinaryComparator binaryComparator = new BinaryComparator(Bytes.toBytes("1500100010"));
//            RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.LESS_OR_EQUAL, binaryComparator);

            //新版本写法：RowFilter(final CompareOperator op, final ByteArrayComparable rowComparator)
            //操作符专门使用了一个枚举类CompareOperator
            RowFilter rowFilter = new RowFilter(CompareOperator.LESS_OR_EQUAL, binaryComparator);

            //由scan设置过滤器
            scan.setFilter(rowFilter);

            ResultScanner resultScanner = students2.getScanner(scan);
            MyHBaseTools.printResultScanner(resultScanner);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 需求2：通过FamilyFilter与SubstringComparator查询列簇名包含in的所有列簇下面的数据
     */
    @Test
    public void scanAndFamilyFilterAndSubstringComparator() {
        try {
            TableName tableName = TableName.valueOf("students2");

            Table students2 = conn.getTable(tableName);

            //创建Scan对象
            Scan scan = new Scan();

            //创建包含比较器
            SubstringComparator substringComparator = new SubstringComparator("nf");

            //创建列簇过滤器
            //FamilyFilter(final CompareOperator op, final ByteArrayComparable familyComparator)
            FamilyFilter familyFilter = new FamilyFilter(CompareOperator.EQUAL, substringComparator);

            //设置过滤器
            scan.setFilter(familyFilter);

            ResultScanner resultScanner = students2.getScanner(scan);

            MyHBaseTools.printResultScanner(resultScanner);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 列名过滤器
     * 需求3：通过QualifierFilter与SubstringComparator查询列名包含ge的列的值
     */
    @Test
    public void scanQualifierFilterAndSubstringComparator() {
        try {
//            TableName tableName = TableName.valueOf("students");
//            Table students2 = conn.getTable(tableName);

            Table students = MyHBaseTools.getTableExample(conn, "students");

            //创建Scan对象
            Scan scan = new Scan();

            //创建包含比较器
            SubstringComparator substringComparator = new SubstringComparator("别");

            //创建列名过滤器
            //QualifierFilter(final CompareOperator op, final ByteArrayComparable qualifierComparator)
            QualifierFilter qualifierFilter = new QualifierFilter(CompareOperator.EQUAL, substringComparator);

            scan.setFilter(qualifierFilter);

            ResultScanner resultScanner = students.getScanner(scan);
            MyHBaseTools.printResultScanner(resultScanner);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 需求4：通过ValueFilter与BinaryPrefixComparator过滤出所有的cell中值以 "张" 开头的学生
     * <p>
     * 1500101001,张雷鹏程,23,男,张理科六班
     */
    @Test
    public void scanValueFilterAndBinaryPrefixComparator() {
        try {
            //使用我们自己定义的工具类获取表的实例
            Table students = MyHBaseTools.getTableExample(conn, "students2");

            //创建Scan对象
            Scan scan = new Scan();

            //创建二进制前缀比较器
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator(Bytes.toBytes("文科"));
            //TODO:保留需求：只过滤name这一列值以张开头的数据

            //创建列值过滤器
            //ValueFilter(final CompareOperator valueCompareOp,  final ByteArrayComparable valueComparator)
            ValueFilter valueFilter = new ValueFilter(CompareOperator.EQUAL, binaryPrefixComparator);

            scan.setFilter(valueFilter);

            ResultScanner resultScanner = students.getScanner(scan);

            MyHBaseTools.printResultScanner(resultScanner);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 专用过滤器分类之单列值过滤器
     * <p>
     * 需求1：通过SingleColumnValueFilter与查询文科班所有学生信息
     */
    @Test
    public void scanSingleColumnValueFilterAndBinaryPrefixComparator() {
        try {
            //获取表的实例
            Table students2 = MyHBaseTools.getTableExample(conn, "students2");

            //创建Scan对象
            Scan scan = new Scan();
            //创建SingleColumnValueFilter过滤器
            //SingleColumnValueFilter(final byte [] family,  列簇
            //      final byte [] qualifier,   列名
            //      final CompareOperator op,  操作符
            //      final org.apache.hadoop.hbase.filter.ByteArrayComparable comparator  比较器)
            SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes("detail"),
                    Bytes.toBytes("clazz"),
                    CompareOperator.EQUAL,
                    new BinaryPrefixComparator(Bytes.toBytes("文科")));

            scan.setFilter(singleColumnValueFilter);

            ResultScanner resultScanner = students2.getScanner(scan);

            MyHBaseTools.printResultScanner(resultScanner);


        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /**
     * SingleColumnValueExcludeFilter
     */
    @Test
    public void scanSingleColumnValueExcludeFilterAndBinaryPrefixComparator() {
        try {
            //获取表的实例
            Table students2 = MyHBaseTools.getTableExample(conn, "students2");

            //创建Scan对象
            Scan scan = new Scan();
            //创建SingleColumnValueFilter过滤器
            //SingleColumnValueFilter(final byte [] family,  列簇
            //      final byte [] qualifier,   列名
            //      final CompareOperator op,  操作符
            //      final org.apache.hadoop.hbase.filter.ByteArrayComparable comparator  比较器)
            SingleColumnValueExcludeFilter singleColumnValueExcludeFilter = new SingleColumnValueExcludeFilter(Bytes.toBytes("detail"),
                    Bytes.toBytes("clazz"),
                    CompareOperator.EQUAL,
                    new BinaryPrefixComparator(Bytes.toBytes("文科")));

            scan.setFilter(singleColumnValueExcludeFilter);

            ResultScanner resultScanner = students2.getScanner(scan);

            MyHBaseTools.printResultScanner(resultScanner);


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 行键前缀过滤器
     * 通过PrefixFilter查询以15001001开头的所有前缀的rowkey
     */
    @Test
    public void scanPrefixFilter() {
        try {
            //获取表的实例
            Table students2 = MyHBaseTools.getTableExample(conn, "students2");

            //创建Scan对象
            Scan scan = new Scan();


            //创建行键前缀过滤器
            PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes("15001001"));

            scan.setFilter(prefixFilter);

            ResultScanner resultScanner = students2.getScanner(scan);

            MyHBaseTools.printResultScanner(resultScanner);


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 分页过滤器PageFilter
     * 需求：通过PageFilter查询三页的数据，每页10条
     */
    @Test
    public void scanPageFilter() {
        try {
            //获取表的实例
            Table students2 = MyHBaseTools.getTableExample(conn, "students2");

            //设置几个变量
            //设置页数
            int pageNum = 5;

            //设置每一页的条数
            long pageDataNum = 10L;

            //设置开始行
            String currentRowKey = "";

            //创建分页过滤器
            PageFilter pageFilter = new PageFilter(pageDataNum);
            Scan scan = new Scan();
            scan.setFilter(pageFilter);

            for (int i = 1; i <= pageNum; i++) {
                System.out.println("========================当前是第 " + i + " 页======================");
                ResultScanner resultScanner = students2.getScanner(scan);
//                MyHBaseTools.printResultScanner(scanner);

                for (Result result : resultScanner) {
                    //获取每一行的行键
                    currentRowKey = Bytes.toString(result.getRow()) + 0;
                    //设置下一次读取开始行
                    scan.withStartRow(Bytes.toBytes(currentRowKey));
                    MyHBaseTools.printResult(result);
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 包装过滤器之SkipFilter  跳过的是一行数据
     */
    @Test
    public void scanSkipFilter() {
        try {
            //获取表的实例
            Table students2 = MyHBaseTools.getTableExample(conn, "students2");

            //创建Scan对象
            Scan scan = new Scan();


            // 定义 ValueFilter 过滤器
            Filter filter1 = new ValueFilter(CompareOperator.NOT_EQUAL, new BinaryComparator(Bytes.toBytes("连鸿晖")));
            // 使用 SkipFilter 进行包装
            Filter filter2 = new SkipFilter(filter1);

            scan.setFilter(filter2);

            ResultScanner resultScanner = students2.getScanner(scan);

            MyHBaseTools.printResultScanner(resultScanner);


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 包装过滤器之WhileMatchFilter 直到某一行结束，包括最后一行
     */
    @Test
    public void scanWhileMatchFilter() {
        try {
            //获取表的实例
            Table students2 = MyHBaseTools.getTableExample(conn, "students2");

            //创建Scan对象
            Scan scan = new Scan();


            // 定义 ValueFilter 过滤器
            Filter filter1 = new ValueFilter(CompareOperator.NOT_EQUAL, new BinaryComparator(Bytes.toBytes("连鸿晖")));
            // 使用 SkipFilter 进行包装
            Filter filter2 = new WhileMatchFilter(filter1);

            scan.setFilter(filter2);

            ResultScanner resultScanner = students2.getScanner(scan);

            MyHBaseTools.printResultScanner(resultScanner);


        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     *  组合过滤器：
     *  需求：通过运用4种比较器过滤出姓于，年纪大于23岁，性别为女，且是理科的学生。
     */
    @Test
    public void scanMoreFilter(){
        try {
            //获取表的实例
            Table students2 = MyHBaseTools.getTableExample(conn, "students2");

            //创建Scan对象
            Scan scan = new Scan();

            //过滤出姓于
            BinaryPrefixComparator binaryPrefixComparator1 = new BinaryPrefixComparator(Bytes.toBytes("于"));
            SingleColumnValueFilter filter1 = new SingleColumnValueFilter(Bytes.toBytes("info"),
                    Bytes.toBytes("name"),
                    CompareOperator.EQUAL,
                    binaryPrefixComparator1);

            //年纪大于23岁
            BinaryComparator binaryComparator = new BinaryComparator(Bytes.toBytes("21"));
            SingleColumnValueFilter filter2 = new SingleColumnValueFilter(Bytes.toBytes("info"),
                    Bytes.toBytes("age"),
                    CompareOperator.GREATER,
                    binaryComparator);

            //性别为女
            BinaryComparator binaryComparator2 = new BinaryComparator(Bytes.toBytes("女"));
            SingleColumnValueFilter filter3 = new SingleColumnValueFilter(Bytes.toBytes("detail"),
                    Bytes.toBytes("gender"),
                    CompareOperator.EQUAL,
                    binaryComparator2);

            //理科的学生
            SubstringComparator substringComparator = new SubstringComparator("理科");
            SingleColumnValueFilter filter4 = new SingleColumnValueFilter(Bytes.toBytes("detail"),
                    Bytes.toBytes("clazz"),
                    CompareOperator.EQUAL,
                    substringComparator);


            //设置过滤器 不能设置多次，如果这么做，只有最后一个过滤器生效
//            scan.setFilter(filter1);
//            scan.setFilter(filter2);
//            scan.setFilter(filter3);
//            scan.setFilter(filter4);

            //创建一个FilterList对象（是Filter子类）
            FilterList filterList = new FilterList();
            filterList.addFilter(filter1);
            filterList.addFilter(filter2);
            filterList.addFilter(filter3);
            filterList.addFilter(filter4);

            scan.setFilter(filterList);


            ResultScanner resultScanner = students2.getScanner(scan);

            MyHBaseTools.printResultScanner(resultScanner);


        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 用于释放资源
     */
    @After
    public void close() {
        try {
            if (conn != null) {
                conn.close();
            }

            if (admin != null) {
                admin.close();
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
